package org.apache.rocketmq.proxy.processor;

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.*;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.common.utils.FutureUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.common.ProxyException;
import org.apache.rocketmq.proxy.common.ProxyExceptionCode;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.processor.validator.DefaultTopicMessageTypeValidator;
import org.apache.rocketmq.proxy.processor.validator.TopicMessageTypeValidator;
import org.apache.rocketmq.proxy.service.ServiceManager;
import org.apache.rocketmq.proxy.service.route.AddressableMessageQueue;
import org.apache.rocketmq.remoting.protocol.NamespaceUtil;
import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/17
 * @方法描述：生产者处理器，该处理器的功能非常清晰，那就是处理来自生产者客户端的所有请求，在第十二版本代码中
 * 这个生产者处理器的内容还很简单，只处理了生产者客户端的发送消息请求
 */
public class ProducerProcessor extends AbstractProcessor {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);

    private final ExecutorService executor;

    //主题消息类型校验器，专门用来校验要发送的消息类型是否与主题配置的类型匹配
    private final TopicMessageTypeValidator topicMessageTypeValidator;


    public ProducerProcessor(MessagingProcessor messagingProcessor,
                             ServiceManager serviceManager, ExecutorService executor) {
        super(messagingProcessor, serviceManager);
        this.executor = executor;
        this.topicMessageTypeValidator = new DefaultTopicMessageTypeValidator();
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理生产者客户端发送消息请求的方法
     */
    public CompletableFuture<List<SendResult>> sendMessage(ProxyContext ctx, QueueSelector queueSelector,
                                                           String producerGroup, int sysFlag, List<Message> messageList, long timeoutMillis) {
        CompletableFuture<List<SendResult>> future = new CompletableFuture<>();

        //记录发送消息操作的初始执行时间
        long beginTimestampFirst = System.currentTimeMillis();
        //用来记录被选中的消息要发送的目标队列
        AddressableMessageQueue messageQueue = null;
        try {
            //得到消息列表中的第一条消息
            Message message = messageList.get(0);
            //得到消息所属主题
            String topic = message.getTopic();
            //判断是否启用了消息主题类型检查器
            if (ConfigurationManager.getProxyConfig().isEnableTopicMessageTypeCheck()) {
                //检查消息主题类型检查器是否存在
                if (topicMessageTypeValidator != null) {
                    //判断当前消息主题不是重试主题和死信队列主题，这两个主题是mq系统内置的主题，在这里不必检查
                    if (!NamespaceUtil.isRetryTopic(topic) && !NamespaceUtil.isDLQTopic(topic)) {
                        //得到当前消息主题配置的消息类型
                        TopicMessageType topicMessageType = serviceManager.getMetadataService().getTopicMessageType(ctx, topic);
                        //从消息属性中解析消息类型
                        TopicMessageType messageType = TopicMessageType.parseFromMessageProperty(message.getProperties());
                        //校验消息类型是否匹配
                        topicMessageTypeValidator.validate(topicMessageType, messageType);
                    }
                }
            }
            //使用消息队列选择器选择目标消息队列， this.serviceManager.getTopicRouteService().getCurrentMessageQueueView(ctx, topic)这行代码会得到该主题下
            //所有可路由的消息队列
            messageQueue = queueSelector.select(ctx, this.serviceManager.getTopicRouteService().getCurrentMessageQueueView(ctx, topic));
            //如果没有选中目标队列则抛出异常
            if (messageQueue == null) {
                throw new ProxyException(ProxyExceptionCode.FORBIDDEN, "no writable queue");
            }
            //为每个消息设置一个唯一标识符
            for (Message msg : messageList) {
                MessageClientIDSetter.setUniqID(msg);
            }
            //构建要发送的消息请求头
            SendMessageRequestHeader requestHeader = buildSendMessageRequestHeader(messageList, producerGroup, sysFlag, messageQueue.getQueueId());
            //记录最终选择的目标消息队列
            AddressableMessageQueue finalMessageQueue = messageQueue;
            future = this.serviceManager.getMessageService().sendMessage(ctx, messageQueue, messageList, requestHeader, timeoutMillis)
                    .thenApplyAsync(sendResultList -> {
                        //遍历发送结果列表
                        for (SendResult sendResult : sendResultList) {
                            //获取消息的事务标志
                            int tranType = MessageSysFlag.getTransactionValue(requestHeader.getSysFlag());
                            //如果消息发送成功且是事务消息，并且有事务Id
                            // if (SendStatus.SEND_OK.equals(sendResult.getSendStatus()) && tranType == MessageSysFlag.TRANSACTION_PREPARED_TYPE && StringUtils.isNotBlank(sendResult.getTransactionId())) {
                            //     在这里填充事务相关数据，因为第十二版本代码还不涉及事务消息，所以我就先把这段代码注释了
                            //     fillTransactionData(ctx, producerGroup, finalMessageQueue, sendResult, messageList);
                            // }
                        }
                        //返回发送结果列表
                        return sendResultList;
                    },this.executor)
                    //更新Broker节点故障信息和性能统计
                    .whenComplete((result, exception) -> {
                        //得到发送消息操作执行完毕的时间
                        long endTimestamp = System.currentTimeMillis();
                        if (exception != null) {
                            //如果消息发送过程中出现了异常，就意味着消息发送失败，那就把这些故障信息更新到MQFaultStrategy对象中
                            //更新得内容包括Broker名称、发送消息操作耗时、broker节点是否需要因为故障被隔离、消息是否成功到达Broker节点
                            this.serviceManager.getTopicRouteService().updateFaultItem(finalMessageQueue.getBrokerName(), endTimestamp - beginTimestampFirst, true, false);
                        } else {
                            //消息发送成功时，也更新相关信息
                            this.serviceManager.getTopicRouteService().updateFaultItem(finalMessageQueue.getBrokerName(),endTimestamp - beginTimestampFirst, false, true);
                        }
                    });
        } catch (Throwable t) {
            future.completeExceptionally(t);
        }
        return FutureUtils.addExecutor(future, this.executor);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：构建发送消息请求的请求头的方法
     */
    protected SendMessageRequestHeader buildSendMessageRequestHeader(List<Message> messageList,
                                                                     String producerGroup, int sysFlag, int queueId) {
        //创建请求头对象
        SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
        //得到要发送的第一条消息
        Message message = messageList.get(0);
        //设置生产者组名称到请求头中
        requestHeader.setProducerGroup(producerGroup);
        //设置消息主题
        requestHeader.setTopic(message.getTopic());
        //设置默认主题，用于自动创建主题
        requestHeader.setDefaultTopic(TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC);
        //设置默认主题的队列数
        requestHeader.setDefaultTopicQueueNums(4);
        //设置消息将要发送到的队列的Id
        requestHeader.setQueueId(queueId);
        //设置系统标志，包含了消息类型等信息
        requestHeader.setSysFlag(sysFlag);
        try {
            //尝试从消息属性中获取消息创建时间并设置到请求头
            requestHeader.setBornTimestamp(Long.parseLong(message.getProperty(MessageConst.PROPERTY_BORN_TIMESTAMP)));
        } catch (Exception e) {
            log.warn("parse born time error, with value:{}", message.getProperty(MessageConst.PROPERTY_BORN_TIMESTAMP));
            //发生异常则使用当前系统时间作为消息创建时间
            requestHeader.setBornTimestamp(System.currentTimeMillis());
        }
        //设置消息标志位，包含了是否压缩、是否事务消息等标志
        requestHeader.setFlag(message.getFlag());
        //将消息自定义属性转换为字符串格式，并且设置到请求头中
        requestHeader.setProperties(MessageDecoder.messageProperties2String(message.getProperties()));
        //初始化消息重试次数为0
        requestHeader.setReconsumeTimes(0);
        //判断是否为批量消息，如果消息列表中包含多条消息，则设置批量标志
        if (messageList.size() > 1) {
            requestHeader.setBatch(true);
        }
        //处理重试主题的特殊逻辑，以下内容在第十二版本代码中还不需要，就算有代码注释，大家也不知道消息重试的作用是什么
        //所以大家可以暂时忽略下面的代码，等后面实现具体功能的时候，大家就清楚了，现在也就是先混个眼熟
        if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
            //获取消息的重试次数
            String reconsumeTimes = MessageAccessor.getReconsumeTime(message);
            if (reconsumeTimes != null) {
                //设置重试次数到请求头中
                requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
                //清除消息属性中的重试次数
                MessageAccessor.clearProperty(message, MessageConst.PROPERTY_RECONSUME_TIME);
            }
            //得到消息最大重试次数
            String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(message);
            if (maxReconsumeTimes != null) {
                //设置最大重试次数到请求头中
                requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
                //清除消息属性中的最大重试次数
                MessageAccessor.clearProperty(message, MessageConst.PROPERTY_MAX_RECONSUME_TIMES);
            }
        }
        return requestHeader;
    }
}